package com.heima.search.listen;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.ArticleConstants;
import com.heima.common.constants.WemediaConstants;
import com.heima.common.exception.CustomException;
import com.heima.feign.article.IArticleClient;
import com.heima.model.article.pojos.ApArticleContent;
import com.heima.model.common.dtos.ResponseResult;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.search.vos.SearchArticleVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

@Component
@Slf4j
@SuppressWarnings("all")
public class SyncArticleListener {
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    @Autowired
    private IArticleClient iArticleClient;


    /**
     * 监听kafka消息，添加文章到es
     *
     * @param message
     */
    @KafkaListener(topics = ArticleConstants.ARTICLE_ES_SYNC_TOPIC)
    public void onMessage(String message) {
        log.info("SyncArticleListener message:{} ", message);
        //1、非空判断
        try {
            if (StringUtils.isNotBlank(message)) {
                //2、解析数据
                SearchArticleVo searchArticleVo = JSON.parseObject(message, SearchArticleVo.class);
                //3、创建IndexRequest对象
                IndexRequest indexRequest = new IndexRequest(ArticleConstants.ES_INDEX);
                //4、将文章id作为es存储id
                indexRequest.id(searchArticleVo.getId().toString());
                //5、构建DSL 直接放message，因为他需要一个json的对象
                indexRequest.source(message, XContentType.JSON);
                //6、发送消息
                restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("sync es error:{} ", e);
        }
    }


    /**
     * 监听文章上下架 对应增加或者删除es文章
     *
     * @param message
     */
    @KafkaListener(topics = WemediaConstants.WM_NEWS_UP_OR_DOWN_TOPIC)
    public void listenIsDown(String message) {
        try {
            //1、参数非空校验
            if (StringUtils.isNotBlank(message)) {
                //2、解析成原来的map数据
                Map map = JSON.parseObject(message, Map.class);
                //3、判断上架还是下架
                Object enable = map.get("enable");
                if (enable.equals(0)) {
                    //3、创建DeleteRequest
                    DeleteRequest deleteRequest = new DeleteRequest(ArticleConstants.ES_INDEX);
                    //4、根据文章id下架
                    deleteRequest.id(map.get("articleId").toString());
                    //4、发送请求
                    restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
                } else if (enable.equals(1)) {
                    //5、远程调用的到返回结果
                    System.out.println(map.get("articleId").toString());
                    ResponseResult responseResult = iArticleClient.rackArticleSaveToES(Long.valueOf(map.get("articleId").toString()));
                    if (!responseResult.getCode().equals(200)) {
                        log.error("纳尼！！！！！！！！！");
                        return;
                    }
                    //6、解析结果
                    // SearchArticleVo data = (SearchArticleVo) responseResult.getData();
                    //先序列化成json
                    String json1 = JSON.toJSONString(responseResult.getData());
                    //再反序列化原来对象
                    SearchArticleVo data = JSON.parseObject(json1, SearchArticleVo.class);
                    //7、保存到es
                    //7.1、创建request连接索引库 id
                    IndexRequest indexRequest = new IndexRequest(ArticleConstants.ES_INDEX).id(data.getId().toString());
                    //7.2、序列化成json
                    String json = JSON.toJSONString(data);
                    //7.3、构建DSL
                    indexRequest.source(json, XContentType.JSON);
                    //7.4、发送消息
                    restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
